//go:build !windows

package runtime_server

import (
	"context"
	"fmt"
	"sync"

	"gitee.com/hasika/gotool/taskx"

	"gitee.com/hasika/common-server/conn/inet"
	"gitee.com/hasika/common-server/epoll"
	"gitee.com/hasika/common-server/idispatcher"
	"gitee.com/hasika/common-server/option"
)

type EpollServer struct {
	serviceOption *option.ServiceOption
	listenOption  *option.RemoteOption
	dispatcher    idispatcher.MessageDispatcher
	ctx           context.Context
	pollerMap     map[int]*epoll.Epoll
	pool          *taskx.Pool[any]
	pollerCnt     int
	ls            inet.IListener
	closeOnce     sync.Once
}

func NewEpollServer(ctx context.Context, serviceOption *option.ServiceOption, listenOption *option.RemoteOption, dispatcher idispatcher.MessageDispatcher) *EpollServer {
	num := serviceOption.GetTaskGoroutineNum()
	return &EpollServer{
		serviceOption: serviceOption,
		listenOption:  listenOption,
		pollerCnt:     100,
		pollerMap:     make(map[int]*epoll.Epoll),
		ctx:           ctx,
		pool:          taskx.NewTaskPool[any](ctx, num, 1000, "epoll"),
		dispatcher:    dispatcher,
	}
}

func (m *EpollServer) Run() error {
	defer m.Stop()
	for epollIndex := 0; epollIndex < m.pollerCnt; epollIndex++ {
		poller, err := epoll.MkEpoll()
		if err != nil {
			return err
		}
		m.pollerMap[epollIndex] = poller
		go m.epollWork(m.ctx, poller)
	}
	for {
		conn, err := m.ls.Accept()
		if err != nil {
			return err
		}
		if conn == nil {
			continue
		}
		ctx := inet.NewWriteInTaskPoolCtx(m.pool)
		conn.InitConnection(conn, m.serviceOption.GetConnectionCallback(), true, ctx, m.listenOption.RecordReadWrite())
		m.addConnectionToPoller(conn)
	}
}

func (m *EpollServer) Listen() error {
	ls, err := NewListener(m.ctx, m.listenOption.Network(), fmt.Sprintf("%s:%d", m.listenOption.Host(), m.listenOption.Port()), m.serviceOption)
	if err != nil {
		return err
	}
	m.ls = ls
	return nil
}

func (m *EpollServer) Stop() {
	m.closeOnce.Do(func() {
		if m.ls != nil {
			m.ls.Close()
		}
	})
}

func (m *EpollServer) GetServerOption() *option.ServiceOption {
	return m.serviceOption
}

func (m *EpollServer) addConnectionToPoller(conn inet.Conn) {
	index := conn.GetFd() % (m.pollerCnt)
	poller := m.pollerMap[index]
	poller.Add(conn)
}

func (m *EpollServer) removeConnectionFromPoller(conn inet.Conn) {
	index := conn.GetFd() % (m.pollerCnt)
	poller := m.pollerMap[index]
	poller.Remove(conn)
}

func (m *EpollServer) epollWork(ctx context.Context, epoller *epoll.Epoll) {
	for {
		select {
		case <-ctx.Done():
			return
		default:
			readableConnection, err := epoller.Wait()
			if err != nil {
				Log.Warnf("epoll wait error %+v", err)
				return
			}
			for _, c := range readableConnection {
				m.readFromConn(c)
			}
		}
	}
}

func (m *EpollServer) readFromConn(c inet.Conn) {
	slicer := m.listenOption.GetFrameReader()
	coder := m.serviceOption.GetCoder()
	allPacketBytes, err := slicer.NonBlockSlicePacketBytes(c)
	if err != nil {
		c.CloseConnectionAllCallback("read err " + err.Error())
		return
	}
	m.addConnectionToPoller(c)
	for _, packetByte := range allPacketBytes {
		packet, err := coder.Unmarshal(packetByte)
		if err != nil {
			Log.Errorf("unmarshal packet error %+v", err)
			continue
		}
		m.dispatcher.DispatchMessage(packet, c)
	}

}
